package personal.wang.manager;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import personal.wang.config.DataMockYamlReader;
import personal.wang.constant.ConstantVar;
import personal.wang.util.ParameterUtil;
import personal.wang.util.SystemInfoUtil;
import personal.wang.zookeeper.HAUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.Properties;

/**
 * @className: ZookeeperManager
 * @Description:
 * @Author: wangyifei
 * @Date: 2025/2/20 16:57
 */
public class ZookeeperManager {
    private static Logger logger = LoggerFactory.getLogger(ZookeeperManager.class);
    private WorkerManager workerManager ;
    private RecordGeneratorManager recordGeneratorManager ;

    public ZookeeperManager(WorkerManager workerManager , RecordGeneratorManager recordGeneratorManager){
        this.workerManager = workerManager ;
        this.recordGeneratorManager = recordGeneratorManager ;

    }

    public WorkerManager getWorkerManager() {
        return workerManager;
    }

    public void setWorkerManager(WorkerManager workerManager) {
        this.workerManager = workerManager;
    }

    public RecordGeneratorManager getRecordGeneratorManager() {
        return recordGeneratorManager;
    }

    public void setRecordGeneratorManager(RecordGeneratorManager recordGeneratorManager) {
        this.recordGeneratorManager = recordGeneratorManager;
    }

    public static void main(String[] args) {
        Properties properties = ParameterUtil.fromArgument(args);
        ZookeeperManager zookeeperManager = new ZookeeperManager(null , null);
        String zkQuorum = DataMockYamlReader.getValueAsString(properties.getProperty("path") , ConstantVar.ZOOKEEPER_CONNECTION);
        zookeeperManager.startWorker();
    }

    public void startWorker(){
        String zkQuorum = DataMockYamlReader.getValueAsString(ConstantVar.ZOOKEEPER_CONNECTION);

        if("none".equals(zkQuorum))return;

        CuratorFramework curatorFramework = HAUtils.buildClient();
        curatorFramework.start();
        try {
            curatorFramework.blockUntilConnected();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        String zookeeperHaPath = DataMockYamlReader.getValueAsString(ConstantVar.ZOOKEEPER_HA_PATH);
        NodeCache nodeCache = new NodeCache(curatorFramework, zookeeperHaPath, false);
        try {
            nodeCache.start(true);
        } catch (Exception e) {
            e.printStackTrace();
        }
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                ChildData currentdata = nodeCache.getCurrentData();
                if(Objects.nonNull(currentdata)){
                    String ipPort = StringUtils.toEncodedString(currentdata.getData(), StandardCharsets.UTF_8);
                    String[] split = ipPort.split(":");
                    String ip = split[0];
                    String port = split[1];
                    NetworkManager.reconnectBoss(ip , port);
                }

            }
        });
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public  void startBoss(){
        String zkQuorum = DataMockYamlReader.getValueAsString(ConstantVar.ZOOKEEPER_CONNECTION);

        if("none".equals(zkQuorum))return;

        CuratorFramework curatorFramework = HAUtils.buildClient();
        curatorFramework.start();
        try {
            curatorFramework.blockUntilConnected();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        String zookeeperHaPath = DataMockYamlReader.getValueAsString(ConstantVar.ZOOKEEPER_HA_PATH);
        String clusterNodeId = DataMockYamlReader.getValueAsString(ConstantVar.CLUSTER_NODE_ID);

        LeaderLatch leaderLatch = new LeaderLatch(curatorFramework, zookeeperHaPath, clusterNodeId);
//        NodeCache nodeCache = new NodeCache(curatorFramework , zookeeperHaPath);
        leaderLatch.addListener(new LeaderLatchListener() {
            @Override
            public void isLeader() {
                try {
                    curatorFramework.setData().forPath(zookeeperHaPath , SystemInfoUtil.getLocalIP()
                            .concat(":")
                            .concat(DataMockYamlReader.getValueAsString(ConstantVar.NETTY_BOSS_PORT))
                            .getBytes(StandardCharsets.UTF_8)) ;
                } catch (Exception e) {
                    e.printStackTrace();
                }
                logger.info("boss " + clusterNodeId + "  get leader");
                // 1. 加载未完成的任务。包含的信息为 状态 模板
                recordGeneratorManager.loadTask(curatorFramework);
                // 2. 清理任务,时间戳超过阈值。
                recordGeneratorManager.clearTask(curatorFramework);
                // 3. 加载 worker, 更新 worker 的状态。
                workerManager.loadWorkerInfoFromZookeeper(curatorFramework);
            }

            @Override
            public void notLeader() {
                logger.info("boss " + clusterNodeId + " does not get leader");
            }
        });
        try {
            leaderLatch.start();
        } catch (Exception e){
            e.printStackTrace();
        }
        logger.info("has leadership:{}" , leaderLatch.hasLeadership());
    }
    public void close(){

    }
}
